package com.examples.mysql2mysql.cdc_sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class CDCSqlTestTable {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);
        env.enableCheckpointing(5000);

        final StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        tEnv.executeSql("CREATE DATABASE IF NOT EXISTS f_saas_cfg");
        // 动态表，此为source表
        tEnv.executeSql("CREATE TABLE f_saas_cfg.test_table (\n" +
                "    id                           STRING,\n" +
                "    name                       STRING,\n" +
                "    del           STRING,\n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "  'connector' = 'mysql-cdc',\n" +
                "  'hostname' = '127.0.0.1',\n" +
                "  'port' = '3306',\n" +
                "  'username' = 'root',\n" +
                "  'password' = '123456',\n" +
                "  'database-name' = 'f_saas_cfg',\n" +
                "  'table-name' = 'test_table',\n" +
                "  'scan.incremental.snapshot.enabled' = 'false'\n" +
                ")");

        // 动态表，此为sink表。sink表和source表的connector不一样
        tEnv.executeSql("CREATE TABLE f_saas_cfg.test_table01 (\n" +
                "    id                           STRING,\n" +
                "    name                       STRING,\n" +
                "    del           STRING,\n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "  'connector' = 'jdbc',\n" +
                "  'url' = 'jdbc:mysql://127.0.0.1:3306/f_saas_cfg',\n" +
                "  'username' = 'root',\n" +
                "  'password' = '123456',\n" +
                "  'table-name' = 'test_table01'\n" +
                ")");

        tEnv.executeSql("INSERT INTO f_saas_cfg.test_table01 (id, name, del)\n" +
                "select f.id,\n" +
                "       f.name,\n" +
                "       f.del\n" +
                "from f_saas_cfg.test_table f\n" +
                "where f.id is not null\n" +
                "  and f.del = '0'");

    }
}
